package com.zpa.redisstreamlistener.processor;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.*;
import org.redisson.api.stream.StreamCreateGroupArgs;
import org.redisson.api.stream.StreamReadGroupArgs;
import org.springframework.core.env.Environment;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.function.Consumer;

/**
 * redis Stream 消费者
 */
@Slf4j
public class RedisStreamConsumer {

    private final RedissonClient redisson;

    private final static String LOCK = "redis:stream:lock:2dfe3f31";

    /**
     * 可以不用配置 但最好配置一下 可以更好的追踪
     * 广播模式得到的消费组 可以保存在机器上 然后通过 JVM 命令修改
     * 启动命令：java -Dredisson.cluster.stream.broadcast-group-id=*** -jar ***.jar
     */
    private final String BROADCAST_GROUP_ID;

    /**
     * 可以不用配置 但最好配置一下 可以更好的追踪
     * 消费者名称 每个 JVM 示例应该用不同的值 主要用于消费竞争
     * 没有被确认（ack）的消息 会存在 PENDING 列表中 通过 redis命令：xpending streamKey groupId 查看 这里面也保存了 消费者名称等信息
     * 启动命令：java -Dredisson.cluster.stream.consumer-name=*** -jar ***.jar
     */
    private final String CONSUMER_NAME;

    public RedisStreamConsumer(RedissonClient redisson) {
        this.redisson = redisson;
        Environment environment = RedisStreamListenerProcessor.applicationContext.getEnvironment();
        this.BROADCAST_GROUP_ID = environment.getProperty("redisson.cluster.stream.broadcast-group-id", "");
        this.CONSUMER_NAME = environment.getProperty("redisson.cluster.stream.consumer-name", "");
    }

    /**
     * 消费者线程 用于查看或者停止
     */
    private static final Map<String, Thread> consumerThreads = new ConcurrentHashMap<>();

    private static final String ip;

    static {
        try {
            ip = InetAddress.getLocalHost().getHostAddress();
        } catch (UnknownHostException e) {
            log.error("获取本机IP地址失败", e);
            throw new RuntimeException(e);
        }
    }

    /**
     * 启动消息处理 自动提交
     * topic 主题
     * groupId 消费者组 相同为负载均衡 不同为广播模式
     * messageProcessor 消息处理器
     * StreamMessageId 消息ID
     */
    public void startProcessing(String topic, String groupId, Consumer<Map.Entry<StreamMessageId, Map<String, Object>>> messageProcessor) {
        startProcessing(topic, groupId, true, (message, ack) -> messageProcessor.accept(message));
    }


    /**
     * 启动消息处理
     * topic 主题
     * groupId 消费者组 相同为负载均衡 不同为广播模式
     * autoSubmit 是否自动提交
     * messageProcessor 消息处理器
     * StreamMessageId 消息ID
     * Acknowledgment 自动提交类
     */
    public void startProcessing(String topic, String groupId, boolean autoSubmit, BiConsumer<Map.Entry<StreamMessageId, Map<String, Object>>, Acknowledgment> messageProcessor) {
        final String finalGroupId = this.judgmentGroupId(groupId);

        // consumerName 消费者名称 主要用于负载均衡
        String consumerName;
        if (StringUtils.isBlank(CONSUMER_NAME)) {
            consumerName = ip + ":" + UUID.randomUUID();
        } else {
            consumerName = ip + ":" + CONSUMER_NAME;
        }
        // 线程标识 主要用于停止
        String threadKey = topic + ":" + groupId + ":" + consumerName;
        if (consumerThreads.containsKey(threadKey)) {
            log.warn("Consumer already running for: {}", threadKey);
            return;
        }
        Thread consumerThread = new Thread(() ->
                processMessages(topic, finalGroupId, consumerName, autoSubmit, messageProcessor), "stream-consumer-" + threadKey);
        consumerThread.setDaemon(true);
        consumerThread.start();
        // 保存线程 用于查看或者停止
        consumerThreads.put(threadKey, consumerThread);
    }

    /**
     * 判断是否为广播模式
     */
    private String judgmentGroupId(String groupId) {
        // 如果不为空 则为负载均衡模式
        if (StringUtils.isNotBlank(groupId)) {
            return groupId;
        }
        // 如果为空 则为广播模式
        if (StringUtils.isBlank(BROADCAST_GROUP_ID)) {
            return ip + ":" + UUID.randomUUID();
        }
        return BROADCAST_GROUP_ID;
    }

    /**
     * 监听并处理消息
     */
    private void processMessages(String topic, String groupId,
                                 String consumerName, boolean autoSubmit, BiConsumer<Map.Entry<StreamMessageId, Map<String, Object>>, Acknowledgment> processor) {
        // 获取Redisson的RStream对象
        RStream<String, Object> stream = redisson.getStream(topic);


        // 创建消费者组（如果不存在）
        try {
            stream.createGroup(StreamCreateGroupArgs
                    .name(groupId)
                    .makeStream());
        } catch (Exception e) {
            log.debug("Consumer group might already exist: {}", e.getMessage());
        }

        while (!Thread.currentThread().isInterrupted()) {
            try {
                // 批量读取 10条 30秒超时
                Map<StreamMessageId, Map<String, Object>> messages =
                        stream.readGroup(groupId, consumerName,
                                StreamReadGroupArgs
//                                        .greaterThan(StreamMessageId.AUTO_GENERATED)
                                        .neverDelivered()
//                                        .noAck()
                                        .count(10)
                                        .timeout(Duration.ofSeconds(30L)));

                // 如果超时了还没有数据需要处理 这里就处理没有 ack 的数据
                if (messages == null || messages.isEmpty()) {
                    // 在 pending 不常见的情况下 加锁就有性能问题，这里查询了没有数据就不需要加锁了
                    List<PendingEntry> pendingEntries = stream.listPending(groupId, StreamMessageId.MIN, StreamMessageId.MAX, 100);
                    log.debug("pendingEntries: {}", pendingEntries);
                    if (pendingEntries == null || pendingEntries.isEmpty()){
                        continue;
                    }
                    RLock lock = redisson.getLock(LOCK);
                    try {
                        // 加锁 保证 只有一个实例 能消费到 为确认数据
                        if (lock.tryLock(1000, TimeUnit.MILLISECONDS)) {
                            // 处理没确认的数据 就是没有 ack 的 pending 数据
                            messages = stream.pendingRange(groupId, StreamMessageId.MIN, StreamMessageId.MAX, 100);
                        }
                    } finally {
                        // 确保锁被释放
                        if (lock.isHeldByCurrentThread()) {
                            lock.unlock();
                        }
                    }
                    if (messages == null || messages.isEmpty()) {
                        continue;
                    }
                }

                // 处理消息
                for (Map.Entry<StreamMessageId, Map<String, Object>> entry : messages.entrySet()) {
                    // 获取消息ID
                    StreamMessageId messageId = entry.getKey();
                    // 获取消息
                    Map<String, Object> message = entry.getValue();
                    try {
                        log.debug("开始处理消息: messageId={} message={}", messageId, message);

                        // 可以手动提交
                        Acknowledgment acknowledgment = new ConsumerAcknowledgment(stream, groupId, messageId);
                        // 处理消息
                        processor.accept(entry, acknowledgment);
                        // 是否手动提交
                        if (autoSubmit) {
                            // 确认消息
                            stream.ack(groupId, messageId);
                        }
                        log.debug("消息处理完成: messageId={}", messageId);
                    } catch (Exception e) {
                        log.error("处理消息失败: messageId={}", messageId, e);
                    }
                }
            } catch (Exception e) {
                log.error("处理消息时发生错误", e);
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException ex) {
                    throw new RuntimeException(ex);
                }
            }
        }
    }

    /**
     * 停止监听
     */
    public void stopProcessing(String topic, String groupId, String consumerName) {
        String threadKey = topic + ":" + groupId + ":" + consumerName;
        Thread thread = consumerThreads.remove(threadKey);
        if (thread != null) {
            thread.interrupt();
            log.debug("Stopped consumer: {}", threadKey);
        }
    }

    /**
     * 停止所有监听
     */
    public void stopAll() {
        consumerThreads.forEach((key, thread) -> thread.interrupt());
        consumerThreads.clear();
    }

    public Map<String, Thread> getConsumerThreads() {
        return RedisStreamConsumer.consumerThreads;
    }


    /**
     * 确认消息
     */
    private static class ConsumerAcknowledgment implements Acknowledgment {
        private final RStream<String, Object> stream;
        private final String groupId;
        private final StreamMessageId messageId;

        public ConsumerAcknowledgment(RStream<String, Object> stream, String groupId, StreamMessageId messageId) {
            this.stream = stream;
            this.groupId = groupId;
            this.messageId = messageId;
        }

        public void acknowledge() {
            stream.ack(groupId, messageId);
        }

    }
}